#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Datetime: 2020/7/28 14:17
# @Author  : CHEN Wang
# @Site    :
# @File    : param_optimization.py
# @Software: PyCharm

"""
脚本说明： 回测组合的参数优化
"""

from quant_researcher.quant.factors.indicator_tools.indicators_util import macd_signal, \
    kdj_signal, rsi_signal, wr_signal, bband_signal, ema_cross_signal, ma_cross_signal
from quant_researcher.quant.performance_attribution.core_functions.performance_analysis.performance import *
from quant_researcher.quant.project_tool.db_operator.db_conn import get_stock_conn, \
    get_derivative_data_conn
import matplotlib.pyplot as plt
import itertools
plt.rcParams['font.sans-serif'] = ['SimHei']  # 用来正常显示中文
plt.rcParams['axes.unicode_minus'] = False  # 用来正常显示负号
con = get_derivative_data_conn()
con_astk = get_stock_conn()


def single_param_opt(quote, single_method, param_name=None, min_bound=5, max_bound=100, step=5, njobs=-1):
    param_list = range(min_bound, max_bound, step)
    print(f'{single_method.__name__}指标参数优化，一共有{len(param_list)}个参数，开始遍历……')

    def inner(param):
        kwarg = {param_name: param}
        print(single_method.__name__, kwarg)
        signal = to_signal(quote, single_method, **kwarg)
        # signal = quote.groupby(level=0, as_index=False).apply(single_method, **kwarg)
        # signal = signal.reset_index(level=0, drop=True).unstack(level=0)
        equity, sharpe_ratio, trades, drawdown = run(quote, signal)
        return param, equity, sharpe_ratio, trades, drawdown

    out = Parallel(n_jobs=njobs, verbose=10)(delayed(inner)(param) for param in param_list)
    param, equitys, sharpe_ratios, tradess, drawdowns = zip(*out)
    df_index = pd.Index(param,name='param')
    equity_df = pd.DataFrame(list(equitys), index=df_index)
    sharpe_df = pd.DataFrame(list(sharpe_ratios), index=df_index)
    trade_df = pd.DataFrame(list(tradess), index=df_index)
    maxdraw_df = pd.DataFrame(list(drawdowns), index=df_index)
    return equity_df, sharpe_df, trade_df, maxdraw_df


def multi_param_opt(quote, method, min_bound=5, max_bound=100, step=5, njobs=-1, **params):
    param_short = list(range(min_bound, max_bound, step))
    param_long = list(range(min_bound*2, max_bound*2, step))
    all_comb = pd.Series(list(itertools.product(param_short, param_long)))
    valid = all_comb.apply(lambda x: (x[-1]>x[-2]*1.5) & (x[-1]<x[-2]*8))
    comb = all_comb[valid].tolist()
    print(f'{method.__name__}指标参数优化，一共有{len(comb)}个参数组合，开始遍历……')

    def inner(param):
        kwarg = {name:value for name, value in zip(params.values(),param)}
        print(method.__name__, kwarg)
        signal = to_signal(quote, method, **kwarg)
        # signal = quote.groupby(level=0, as_index=False).apply(method, **kwarg)
        # signal = signal.reset_index(level=0, drop=True).unstack(level=0).ffill()
        equity, sharpe_ratio, trades, drawdown = run(quote, signal)
        return param, equity, sharpe_ratio, trades, drawdown

    out = Parallel(n_jobs=njobs, verbose=10)(delayed(inner)(param) for param in comb)
    param, equitys, sharpe_ratios, tradess, drawdowns = zip(*out)
    df_index = pd.MultiIndex.from_tuples(param, names=['short','long'])
    equity_df = pd.DataFrame(list(equitys), index=df_index)
    sharpe_df = pd.DataFrame(list(sharpe_ratios), index=df_index)
    trade_df = pd.DataFrame(list(tradess), index=df_index)
    maxdraw_df = pd.DataFrame(list(drawdowns), index=df_index)
    return equity_df, sharpe_df, trade_df, maxdraw_df


def result_stats(result, indicator, period_length, valid_trade=48):
    equity_df, sharpe_df, trade_df, maxdraw_df = result
    valid_turnover = trade_df <= round(period_length)*valid_trade
    v_sharpe = sharpe_df[valid_turnover]
    v_equity = equity_df[valid_turnover]
    # v_annual_ret = v_equity **(1/period_length)-1
    v_annual_ret = v_equity.stack().rename('annual_ret')
    v_sharpe = v_sharpe.stack().rename('sharpe')
    out = pd.concat([v_annual_ret,v_sharpe],axis=1)
    out['indicator'] = indicator
    out = out.set_index('indicator', append=True).dropna(subset=['sharpe'])
    mask = out.groupby('sec_code')['sharpe'].idxmax()
    temp = out.loc[mask]
    return temp


def to_signal(quote, method, **kwargs):
    signal = quote.groupby(level=0, as_index=False).apply(method, **kwargs)
    signal = signal.reset_index(level=0, drop=True).unstack(level=0).ffill()
    return signal


def to_equity(quote, signal):
    open_p = quote['open'].unstack(level=0).ffill()
    ret = open_p.pct_change()
    equity = ret.where((signal > 0).shift(2), 0)
    equity = (equity + 1).cumprod()
    return equity


def run(quote, signal):
    open_p = quote['open'].unstack(level=0).ffill()
    ret = open_p.pct_change()
    # bench_equity = (ret+1).cumprod()
    equity = ret.where((signal > 0).shift(2),0).where((signal > 0).shift(1),0)
    sharpe_ratio = equity.apply(sharpe_ratio)
    equity = (equity + 1).cumprod()
    annual_ret = equity.iloc[-1] ** (245/open_p.count())-1
    trades = signal.apply(trade_count)
    drawdown = equity.apply(max_drawdown)
    return annual_ret, sharpe_ratio, trades, drawdown


def multi_param_to_single(multi_stats):
    temp = multi_stats.reset_index(level=[0,1,3])
    temp['param'] = temp['short'].astype(str) + ',' + temp['long'].astype(str)
    temp = temp.set_index(['param','indicator'], append=True).drop(columns=['long','short']).reorder_levels(['param', 'sec_code', 'indicator'])
    return temp


class BackTesting(object):
    def __init__(self, begin_date='', end_date='', target_list=None):
        self.begin_date = begin_date
        self.end_date = end_date
        self.targets = target_list
        self.quote = self.get_quote()

    def get_quote(self):
        if len(self.targets)==1:
            tt = f'({self.targets[0]})'
        else:
            tt = tuple(self.targets)
        quote = pd.read_sql("select sec_code, trade_date, open, high, low, close "
                            "from tk_astkdb.astk_qt_stockdailyquote "
                            f"where sec_code in {tt} "
                            f"and trade_date >= '{self.begin_date}' "
                            f"and trade_date <= '{self.end_date}' "
                            f"order by sec_code, trade_date ", con_astk)
        quote = quote.set_index(['sec_code','trade_date'])
        return quote


if __name__ == '__main__':
    period_start, period_end = '20160101','20200515'
    period_length = (pd.to_datetime(period_end) - pd.to_datetime(period_start)).days / 365
    target_name = pd.read_sql("SELECT fund_code, fund_sname from mf_bd_fndinfo "
                         "where operation_way='6023' ORDER BY fund_code", con)
    # target = ['159911','159901']
    target = target_name['fund_code'].tolist()
    test = BackTesting(period_start,period_end,target)
    result_rsi = single_param_opt(test.quote, single_method=rsi_signal, param_name='timeperiod', step=3, min_bound=5)
    stats_rsi = result_stats(result_rsi, 'rsi', period_length)

    result_kdj = single_param_opt(test.quote, njobs=-1, single_method=kdj_signal, param_name='fastk_period', step=3, min_bound=9)
    stats_kdj = result_stats(result_kdj, 'kdj', period_length)

    result_wr = single_param_opt(test.quote, single_method=wr_signal, param_name='timeperiod', step=3, min_bound=5)
    stats_wr = result_stats(result_wr, 'wr', period_length)

    result_bband = single_param_opt(test.quote, single_method=bband_signal, param_name='timeperiod', step=4, min_bound=20)
    stats_bband = result_stats(result_bband, 'bband',period_length)

    result_ma = multi_param_opt(test.quote, njobs=-1, method=ma_cross_signal, step=5, param1='short_period', param2='long_period')
    stats_ma = result_stats(result_ma, 'ma',period_length)

    result_ema = multi_param_opt(test.quote, method=ema_cross_signal, step=5, param1='short_period', param2='long_period')
    stats_ema = result_stats(result_ema, 'ema',period_length)

    result_macd = multi_param_opt(test.quote, method=macd_signal, step=5, min_bound=12, param1='fastperiod', param2='slowperiod')
    stats_macd = result_stats(result_macd, 'macd',period_length)

    stats_ma = multi_param_to_single(stats_ma)
    stats_ema = multi_param_to_single(stats_ema)
    stats_macd = multi_param_to_single(stats_macd)
    output = pd.concat([stats_rsi, stats_kdj, stats_bband,stats_wr,stats_ma, stats_ema, stats_macd])
    output.index = output.index.set_levels(output.index.levels[0].astype(str),level=0)
    output = output.reset_index()
    # best = output.idxmax()
    # best = best[best.notna()]
    # out = pd.DataFrame(best.tolist(), columns=['param','indicator'],index=best.index)
    outcome = output.merge(target_name, left_on='sec_code',right_on='fund_code')
    index = pd.read_sql("select fund_code, index_code "
                        "from tk_mfdb_23.mf_bd_perftracktargets "
                        f"where fund_code in {tuple(outcome.fund_code)} "
                        f"and end_date is null "
                        f"and index_code is not null", con)
    outcome = outcome.merge(index, on='fund_code')
    # fund_macd = outcome[outcome.indicator=='macd'].sort_values(by='index_code')
    # fund_rsi = outcome[outcome.indicator=='rsi'].sort_values(by='index_code')
    # fund_bband = outcome[outcome.indicator=='bband'].sort_values(by='index_code')
    # fund_kdj = outcome[outcome.indicator=='kdj'].sort_values(by='index_code')
    # fund_ema = outcome[outcome.indicator=='ema'].sort_values(by='index_code')
    # fund_ma = outcome[outcome.indicator=='ma'].sort_values(by='index_code')
    # fund_wr = outcome[outcome.indicator=='wr'].sort_values(by='index_code')


    # ss = to_signal(test.quote, kdj, fastk_period=27)
    # result = run(test.quote,ss)
    # equity_df, sharpe_df, trade_df, maxdraw_df = ma_result

